package com.jds.test.proxy;

import com.jds.test.proto.ClientHandler4Heart;
import com.jds.test.proto.ClientHeartbeatHandler;
import com.jds.test.proto.MyBaseProto;
import com.sun.org.apache.bcel.internal.generic.INEG;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.internal.ConcurrentSet;

import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Created by sunyuming on 18/8/1.
 */
public class Client4Reconnect  {

    /**
     * 可用的
     */
    public static ConcurrentSet<String> setAvailableServers = new ConcurrentSet<>();

    /**
     * 记录Proxy传递的信息服务清单，及相应的线程，线程可能是正常连接，也可能在重连
     */
    private static ConcurrentHashMap<String, Thread> mapThread = new ConcurrentHashMap();

    private static final String SPLITTER = ":";

    /**
     * 外部开启信任某后端服务
     * @param index
     */
    public static void create(String index) {

        Thread thread = new Thread() {
            @Override
            public void run() {
                new Client4Reconnect(index).boot();
            }
        };

        mapThread.put(index, thread);
        System.out.println("***************开启信任线程－" + index + SPLITTER +  thread.getName());
        thread.start();
    }

    /**
     * 外部不再信任某后端服务
     * @param index     */
    public static void interrupt(String index) {
        Thread thread = mapThread.get(index);
        thread.interrupt();
        mapThread.remove(index);
        System.out.println("***************退出信任线程－" + index + SPLITTER + thread.getName());
    }

    /**
     * 内部将自身服务置为可用，对外暴露
     */
    private void addMe() {
        setAvailableServers.add(index);
        System.out.println("--------------新增可用服务" + index + SPLITTER + "ALL - " + setAvailableServers.toString());
    }

    /**
     * 内部将自身服务置为不可用，对外不暴露
     */
    private void removeMe() {
        setAvailableServers.remove(index);
        System.out.println("--------------去除可用服务" + index + SPLITTER + "ALL - " + setAvailableServers.toString());
    }


    private String ipAddress;
    private Integer port;
    private String index;

    private Client4Reconnect(String index) {
        String [] items = index.split(SPLITTER);
        String ipAddress = items[0];
        int port = Integer.valueOf(items[1]);
        this.ipAddress = ipAddress;
        this.port = port;
        this.index = index;
    }

    private void reConnect() {
        connect();
    }

    private void connect() {
        try {
            ChannelFuture futrue = bootstrap.connect(new InetSocketAddress(ipAddress, port)).sync();
            futrue.addListener(new ConnectionListener(this));
            System.out.println("连接成功－" + index);
            addMe();
            futrue.channel().closeFuture().sync();
            System.out.println("连接异常中断，重新连接－" + index);
            removeMe();
            reConnect();
        } catch (InterruptedException e) {
            System.out.println("终止信任，断开连接，服务不再启用－" + index);
            removeMe();
        } catch (Exception e) {
            System.out.println("连接失败－" + index);
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e1) {
                System.out.println("终止信任，不再重连－" + index);
                return;
            }
            System.out.println("重新连接－" + index);
            reConnect();
        } finally {
            worker.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {

        String ipAddress = "127.0.0.1";
        int port = 8866;
        String index = ipAddress + SPLITTER + port;
        create(index);

        Thread.sleep(10000);
        interrupt(index);
    }

    private Bootstrap bootstrap;
    private EventLoopGroup worker;

    private void boot()  {

        worker = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(worker);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();
                pipeline.addLast(new LengthFieldBasedFrameDecoder(10000, 0, 4, 0, 4));
                pipeline.addLast(new ProtobufDecoder(MyBaseProto.BaseProto.getDefaultInstance()));
                pipeline.addLast(new LengthFieldPrepender(4));
                pipeline.addLast(new ProtobufEncoder());

                pipeline.addLast(new IdleStateHandler(61, 30, 0, TimeUnit.SECONDS));
                pipeline.addLast(new ClientHeartbeatHandler());

                pipeline.addLast(new ClientHandler4Heart());
            }
        });

        connect();
    }

    private static class ConnectionListener implements ChannelFutureListener {

        Client4Reconnect client4Reconnect;

        public ConnectionListener(Client4Reconnect client4Reconnect) {
            this.client4Reconnect = client4Reconnect;
        }
        @Override
        public void operationComplete(ChannelFuture channelFuture) throws Exception {

            if (!channelFuture.isSuccess()) {
                System.out.println("Reconnect");
                final EventLoop loop = channelFuture.channel().eventLoop();
                loop.schedule(new Runnable() {
                    @Override
                    public void run() {
                        client4Reconnect.connect();
                    }
                }, 3L, TimeUnit.SECONDS);
            }
        }
    }
}